package org.ala.tcp_feign_client.spring.feign.clients.impl;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.stream.Collectors;

import org.ala.tcp_feign_client.ITcpFeignClient;
import org.ala.tcp_feign_client.context.TcpFeignClientContext;
import org.ala.tcp_feign_client.spring.TcpFeignClientSpringContext;
import org.ala.tcp_feign_client.spring.feign.TcpFeign.LoginSuccessConsumer;
import org.ala.tcp_feign_client.spring.feign.clients.AbstractTcpFeignClientHolder;
import org.ala.tcp_feign_common.logs.LoggerFactory;
import org.ala.tcp_feign_common.msg.biz_resp.BizRespByteMessage;
import org.slf4j.Logger;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.discovery.DiscoveryClient;


/**
 * 从eureka中读取服务端节点信息
 * 
 * @author 骆毅(ala)
 * @date 2024年4月23日
 */
public class EurekaTcpFeignClientHolder extends AbstractTcpFeignClientHolder {
	
	
	static Logger log = LoggerFactory.getLogger("tcp-feign", "client");
	
	
	/**
	 * 服务端metadata-map里需要的数据key
	 */
	public static final String TCP_FEIGN_PORT = "tcp-feign-port",
								TCP_FEIGN_USERNAME = "tcp-feign-username",
								TCP_FEIGN_PASSWORD = "tcp-feign-password";
	
	
	/**
	 * 没30s刷一次eureka
	 * <p>	所以有新的服务注册进来，30s后才能感知到
	 */
	static LinkedList<EurekaTcpFeignClientHolder> holderQueue;
	static {
		holderQueue = new LinkedList<EurekaTcpFeignClientHolder>();
		Executors.newScheduledThreadPool(1).scheduleAtFixedRate(() -> {
			Iterator<EurekaTcpFeignClientHolder> iterator = holderQueue.iterator();
			
			//	要删除的clients
			List<ITcpFeignClient> removes = new ArrayList<>();
			while (iterator.hasNext()) {
				EurekaTcpFeignClientHolder holder = iterator.next();
				List<ITcpFeignClient> rs = holder.refresh();
				if (rs != null && !rs.isEmpty()) {
					removes.addAll(rs);
				}
			}
			//	如果存在需要删除的client，统一删掉
			if (!removes.isEmpty()) {
				removes.forEach(r -> {
					TcpFeignClientContext.instance().close(r.host(), r.port(), r.suffix(), r.username(), r.password());
					
					log.info("[EurekaTcpFeignClientHolder refresh] 下线的节点 host:" + r.host() + " port:" + r.port());
				});
			}
		}, 30, 30, TimeUnit.SECONDS);
	}
	
	
	public EurekaTcpFeignClientHolder(String suffix, Consumer<BizRespByteMessage> respCallback) {
		super(suffix, respCallback);
		
		holderQueue.add(this);
	}
	
	
	/**
	 * 当前持有的服务节点
	 */
	protected Set<ServerInfo> serverInfos;
	/**
	 * 服务名称
	 */
	protected String appName;

	
	@Override
	protected String[] allHosts(String hosts) {
		appName = hosts;
		if (serverInfos == null) {
			serverInfos = loadInstances(appName);
		}
		return serverInfos.stream().map(s -> s.toString()).collect(Collectors.toList()).toArray(new String[0]);
	}
	
	
	/**
	 * ServiceInstance -> String ${host}:${port}:${uername}:${password}
	 */
	protected ServerInfo instanceToInfo(ServiceInstance ins) {
		String host = ins.getHost();
		
		Integer port = null;
		try { port = Integer.parseInt(ins.getMetadata().get(TCP_FEIGN_PORT)); }
		catch (Exception e) {
			log.error("[EurekaTcpFeignClientHolder instanceToHosts] 解析端口失败，该服务端无法使用。请检查该服务端注册的metadata ins:" + ins);
			return null;
		}
		
		String username = ins.getMetadata().get(TCP_FEIGN_USERNAME);
		String password = ins.getMetadata().get(TCP_FEIGN_PASSWORD);
		
		return new ServerInfo(host, port, username, password);
	}
	
	
	/**
	 * 根据appName取服务端信息
	 */
	protected Set<ServerInfo> loadInstances(String appName) {
		DiscoveryClient discoveryClient = TcpFeignClientSpringContext.getBeanByClass(DiscoveryClient.class);
		List<ServiceInstance> ins = discoveryClient.getInstances(appName);
		
		return ins.stream().map(s -> instanceToInfo(s)).filter(s -> s!=null).collect(Collectors.toSet());
	}
	
	
	/**
	 * 重刷客户端，检测是否有新结点加入
	 */
	protected List<ITcpFeignClient> refresh() {
		Set<ServerInfo> ins = loadInstances(appName);
		if (serverInfos == null) {serverInfos = ins; return null;}
		else {
			//	比较新增的节点
			Set<ServerInfo> adds = new HashSet<>(ins);
			adds.removeAll(serverInfos);
			
			//	比较要删除的节点
			Set<ServerInfo> removes = new HashSet<>(serverInfos);
			removes.removeAll(ins);
			
			if (adds.isEmpty() && removes.isEmpty()) {return null;}
			else {
				ArrayList<ITcpFeignClient> newClients = new ArrayList<>();
				ArrayList<ITcpFeignClient> removeClients = new ArrayList<>();
				//	删掉老的
				if (!removes.isEmpty()) {
					for (int i = 0 ; i < clients.size() ; i++) {
						ITcpFeignClient client = clients.get(i);
						ServerInfo info = new ServerInfo(client.host(), client.port(), client.username(), client.password());
						if (removes.contains(info)) {
							removeClients.add(client);
						} else {
							newClients.add(client);
						}
					}
				}
				//	追加新的链接
				if (!adds.isEmpty()) {
					adds.forEach(info -> {
						newClients.add(createClient(info.toString()));
						
						log.info("[EurekaTcpFeignClientHolder refresh] 检测到新追加的节点 info:" + info);
					});
				}
				//	给新clients列表追加登录回调
				for (int i = 0 ; i < newClients.size() ; i++) {
					newClients.get(i).bindLoginCallback(new LoginSuccessConsumer(i, newClients));
				}
				//	切换clients
				clients = newClients;
				//	删除老的连接（统一删除）
				return removeClients;
			}
		}
	}
	public void initClients(String hosts, ArrayList<ITcpFeignClient> clients) {
		String[] remoteHosts = allHosts(hosts);
		
		//	初始化链接
		for (String hostStr : remoteHosts) {
			appendClient(hostStr, clients);
		}
	}
	/**
	 * 追加client
	 * @param	hostStr: ${host}:${port}:${username}:${password}
	 */
	protected void appendClient(String hostStr, ArrayList<ITcpFeignClient> clients) {
		ITcpFeignClient client = createClient(hostStr);
		
		//	拿到其索引
		int i = clients.size();
		clients.add(client);
		client.bindLoginCallback((c) -> new LoginSuccessConsumer(i, clients));
	}
	
	
	/**
	 * 节点信息
	 */
	static class ServerInfo {
		String host;
		int port;
		String username;
		String password;
		
		public ServerInfo(String host, int port, String username, String password) {
			super();
			this.host = host;
			this.port = port;
			this.username = username;
			this.password = password;
		}
		
		public int hashCode() {
			return host.hashCode() + port + username.hashCode() + password.hashCode();
		}
		public boolean equals(Object obj) {
			if (!(obj instanceof ServerInfo)) {return false;}
			ServerInfo info = (ServerInfo) obj;
			return info.host.equals(this.host)
					&& info.port == this.port
					&& info.username.equals(this.username)
					&& info.password.equals(this.password);
		}
		public String toString() {
			return String.format("%s:%d:%s:%s", host, port, username, password);
		}
	}
}
